package com.rem.flink.flink5Watermark;

import com.rem.flink.flink2Source.ClickSource;
import com.rem.flink.flink2Source.Event;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.util.HashSet;

/**
 * 增量聚合函数（AggregateFunction）reduceFunction 可以解决大多数规约聚合问题，但是聚合状态的类型，输出结果的类型必须和输入的类型一致
 * 所以使用聚合函数AggregateFunction 可以实现类型不同
 *
 *
 * <p>
 * 设置窗口    滚动窗口    是一种对数据进行“均匀切片”的划分方式
 * 滑动窗口    窗口之间并不是首尾相接的，而是可以“错开”一定的位置。如果看作一个窗口的运动
 * 会话窗口    数据来了之后就开启一个会话窗口，如果接下来还有数据陆续到来，那么就一直保持会话；如果一段时间一直没收到数据，那就认为会话超时失效，窗口自动关闭。
 * 全局会话    就是全局窗口的意思，将所有的key分在一个窗口
 * <p>
 * 窗口分配器
 * 滚动处理时间窗口 TumblingProcessingTimeWindows.of(Time.seconds(5))
 * 滑动处理时间窗口 SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(5))
 * 滚动事件时间窗口 TumblingEventTimeWindows.of(Time.seconds(5))
 * 滑动事件时间窗口 SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5))
 * 处理时间会话窗口 ProcessingTimeSessionWindows.withGap(Time.seconds(10))      或者.withDynamicGap()
 * 事件时间会话窗口 EventTimeSessionWindows.withGap(Time.seconds(10))
 * <p>
 * <p>
 * 计数窗口
 * 滚动计数窗口   .keyBy(...).countWindow(10)
 * 滑动计数窗口   .keyBy(...).countWindow(10，3)
 *
 * <p>
 * 统计同一分区PV和UV, 再相除 得pv平均访问量
 *
 * @author Rem
 * @date 2022/10/11
 */
public class WindowAggregateTest {
    public static void main(String[] args) throws Exception {
        //1.创建执行环境
        StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.getExecutionEnvironment();
        //2.从socket中读取流
        DataStreamSource<Event> stream = streamEnv.addSource(new ClickSource());

        // 插入水位线的逻辑
        stream.assignTimestampsAndWatermarks(
                        //插入水位线 针对乱序设置延迟时间  顺序不需要
                        WatermarkStrategy.<Event>forMonotonousTimestamps()
                                //提取时间戳
                                .withTimestampAssigner((element, recordTimestamp) -> element.getTimestamp())).keyBy(Event::getUser)
                //设置滚动事件窗口
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                /**
                 * 聚合 统计同一分区PV和UV, 再相除 得pv平均访问量
                 */
                .aggregate(new AggregateFunction<Event, Tuple2<HashSet<String>, Long>, Double>() {
                    /**
                     * 创建累加器
                     * @return
                     */
                    @Override
                    public Tuple2<HashSet<String>, Long> createAccumulator() {
                        return Tuple2.of(new HashSet<>(), 0L);
                    }

                    /**
                     * 数据每来一次就加一次并返回给累加器
                     * @param event
                     * @param accumulator
                     * @return
                     */
                    @Override
                    public Tuple2<HashSet<String>, Long> add(Event event, Tuple2<HashSet<String>, Long> accumulator) {
                        accumulator.f0.add(event.getUser());
                        return Tuple2.of(accumulator.f0, accumulator.f1 + 1L);
                    }

                    /**
                     * 窗口关闭时，增加聚合函数，将结果发送到下游
                     * @param accumulator
                     * @return
                     */
                    @Override
                    public Double getResult(Tuple2<HashSet<String>, Long> accumulator) {
                        return (double) (accumulator.f1 / accumulator.f0.size());
                    }

                    /**
                     * 合并两个累加器，返回具有合并状态的累加器 (合并场景下才会被调用 例如会话窗口)
                     * @param a
                     * @param b
                     * @return
                     */
                    @Override
                    public Tuple2<HashSet<String>, Long> merge(Tuple2<HashSet<String>, Long> a, Tuple2<HashSet<String>, Long> b) {
                        return null;
                    }
                }).print();

        //7.执行流
        streamEnv.execute();

    }
}
